package com.ww.exercise.coroutine.hard.q20

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import java.util.*
import java.util.concurrent.ConcurrentSkipListSet
import java.util.concurrent.atomic.AtomicInteger
import kotlin.math.roundToLong
import kotlin.random.Random


// 请求类型枚举
enum class RequestType { QUERY, INSERT, DELETE }

// 请求数据类（实现Comparable用于排序）
data class PrioritizedRequest(
    val id: String,
    val type: RequestType,
    val isEmergency: Boolean = false,
    val submitTime: Long = System.currentTimeMillis()
) : Comparable<PrioritizedRequest> {
    // 排序规则：紧急请求优先，相同优先级按提交时间升序
    override fun compareTo(other: PrioritizedRequest): Int {
        return when {
            this.isEmergency && !other.isEmergency -> -1 // 本请求紧急，优先级更高
            !this.isEmergency && other.isEmergency -> 1 // 其他请求紧急，优先级更高
            else -> this.submitTime.compareTo(other.submitTime) // 按提交时间排序
        }
    }
}

// 结果统计数据类
data class RequestResult(
    val request: PrioritizedRequest,
    val waitTime: Long, // 等待连接的时间ms
    val processTime: Long // 处理时间ms
)

// 连接池隔离管理器
class ConnectionIsolationManager {
    // 连接配额设置
    private val querySemaphore = Semaphore(30) // 查询最多30个连接
    private val insertSemaphore = Semaphore(15) // 新增最多15个连接
    private val deleteSemaphore = Semaphore(5) // 删除最多5个连接

    // 线程安全的优先级队列（使用ConcurrentSkipListSet实现排序）
    private val priorityQueue = ConcurrentSkipListSet<PrioritizedRequest>()

    // 处理通道（用于传递排序后的请求）
    private val processingChannel = Channel<PrioritizedRequest>(Channel.UNLIMITED)

    // 统计信息存储
    private val queryResults = mutableListOf<RequestResult>()
    private val insertResults = mutableListOf<RequestResult>()
    private val deleteResults = mutableListOf<RequestResult>()

    // 独立协程作用域（按请求类型隔离）
    private val queryScope = CoroutineScope(Dispatchers.IO + CoroutineName("QueryPool"))
    private val insertScope = CoroutineScope(Dispatchers.IO + CoroutineName("InsertPool"))
    private val deleteScope = CoroutineScope(Dispatchers.IO + CoroutineName("DeletePool"))
    private val schedulerScope = CoroutineScope(Dispatchers.IO + CoroutineName("Scheduler"))

    // TODO init使用
    init {
        // 启动调度协程：从优先级队列去任务并发送到处理通道
        schedulerScope.launch {
            while (isActive) {
                // 取出优先级最高的请求（线程安全）
                val nextRequest = priorityQueue.pollFirst()
                if (nextRequest != null) {
                    processingChannel.send(nextRequest)
                } else {
                    delay(1) // 无任务短暂休眠，避免空轮询
                }
            }
        }

        // 启动处理协程：从通道接收请求并处理
        schedulerScope.launch {
            processingChannel.consumeEach { request ->
                // 按请求类型获取对应的作用域和信号量
                val (semaphore, scope) = when (request.type) {
                    RequestType.QUERY -> Pair(querySemaphore, queryScope)
                    RequestType.INSERT -> Pair(insertSemaphore, insertScope)
                    RequestType.DELETE -> Pair(deleteSemaphore, deleteScope)
                }

                // 异步处理请求：先申请连接，再执行逻辑（无join()）
                scope.launch {
                    // 关键：在协程启动后立即申请连接，确保连接被消耗
                    semaphore.withPermit {
                        val waitTime = System.currentTimeMillis() - request.submitTime
                        val processStart = System.currentTimeMillis()

                        try {
                            simulateDatabaseOperation(request.type)
                            val processTime = System.currentTimeMillis() - processStart
                            val result = RequestResult(request, waitTime, processTime)

                            // 线程安全集合，无需同步块
                            when (request.type) {
                                RequestType.QUERY -> queryResults.add(result)
                                RequestType.INSERT -> insertResults.add(result)
                                RequestType.DELETE -> deleteResults.add(result)
                            }

                            printStatisticsIfNeeded()
                        } catch (e: Exception) {
                            println("请求 ${request.id} 处理失败: ${e.message}")
                        }
                    }
                }
            }
        }
    }

    // 提交请求到优先级队列
    suspend fun submitRequest(request: PrioritizedRequest) {
        priorityQueue.add(request)
    }

    // 模拟数据库操作（不同类型耗时不同）
    private suspend fun simulateDatabaseOperation(type: RequestType) {
        val delayMs = when (type) {
            RequestType.QUERY -> Random.nextLong(5, 20)
            RequestType.INSERT -> Random.nextLong(10, 30)
            RequestType.DELETE -> Random.nextLong(15, 40)
        }
        delay(delayMs)
    }

    // 定期打印统计信息
    private fun printStatisticsIfNeeded() {
        // 控制打印频率，避免输出过多
        val queryCount = queryResults.size
        val insertCount = insertResults.size
        val deleteCount = deleteResults.size

        if (queryCount % 500 == 0 || insertCount % 200 == 0 || deleteCount % 100 == 0) {
            val queryAvg = if (queryCount > 0) queryResults.map { it.waitTime }.average().roundToLong() else 0
            val insertAvg = if (insertCount > 0) insertResults.map { it.waitTime }.average().roundToLong() else 0
            val deleteAvg = if (deleteCount > 0) deleteResults.map { it.waitTime }.average().roundToLong() else 0

            println("\n[${Date()}] 统计信息:")
            println("查询请求: $queryCount 个 | 平均等待: ${queryAvg}ms | 可用连接: ${querySemaphore.availablePermits}")
            println("新增请求: $insertCount 个 | 平均等待: ${insertAvg}ms | 可用连接: ${insertSemaphore.availablePermits}")
            println("删除请求: $deleteCount 个 | 平均等待: ${deleteAvg}ms | 可用连接: ${deleteSemaphore.availablePermits}")
        }
    }

    // 关闭资源
    fun shutdown() = runCatching {
        queryScope.cancel()
        insertScope.cancel()
        deleteScope.cancel()
        schedulerScope.cancel()
        processingChannel.close()
        println("\n所有请求处理完成，最终统计：")
        printStatisticsIfNeeded()
    }
}

// 模拟请求生成器
suspend fun generateRequests(manager: ConnectionIsolationManager, durationMs: Long) {
    val endTime = System.currentTimeMillis() + durationMs
    val requestId = AtomicInteger(0)

    coroutineScope {
        // 生成查询请求（QPS 5万）
        launch(Dispatchers.IO) {
            while (System.currentTimeMillis() < endTime) {
                val isEmergency = Random.nextDouble() < 0.10
                val request = PrioritizedRequest(
                    id = "Q-${requestId.incrementAndGet()}",
                    type = RequestType.QUERY,
                    isEmergency = isEmergency
                )
                manager.submitRequest(request)
                delay(20)
            }
        }

        // 生成新增请求（QPS 2万）
        launch(Dispatchers.IO) {
            while (System.currentTimeMillis() < endTime) {
                val request = PrioritizedRequest(
                    id = "I-${requestId.incrementAndGet()}",
                    type = RequestType.INSERT
                )
                manager.submitRequest(request)
                delay(50)
            }
        }

        // 生成删除请求（QPS 1万）
        launch(Dispatchers.IO) {
            while (System.currentTimeMillis() < endTime) {
                val request = PrioritizedRequest(
                    id = "D-${requestId.incrementAndGet()}",
                    type = RequestType.DELETE
                )
                manager.submitRequest(request)
                delay(100)
            }
        }
    }
}

// 主函数
fun main(): Unit = runBlocking {
    val manager = ConnectionIsolationManager()
    println("开始处理请求...")

    generateRequests(manager, 3000000)
    manager.shutdown()
}